package com.atguigu.flink0922.chapter08;

import com.atguigu.flink0922.bean.OrderEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/9 13:53
 */
public class Flink06_High_Project_Order {
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        WatermarkStrategy<OrderEvent> wms = WatermarkStrategy
            .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
            .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                @Override
                public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                    return element.getEventTime();
                }
            });
        env
            .readTextFile("input/OrderLog.csv")
            .map(line -> {
                String[] datas = line.split(",");
                return new OrderEvent(
                    Long.valueOf(datas[0]),
                    datas[1],
                    datas[2],
                    Long.parseLong(datas[3]) * 1000);
                
            })
            .assignTimestampsAndWatermarks(wms)
            .keyBy(OrderEvent::getOrderId)
            .process(new KeyedProcessFunction<Long, OrderEvent, String>() {
                
                private ValueState<OrderEvent> createState;
                private ValueState<OrderEvent> payState;
                
                
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    createState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<OrderEvent>("createState", OrderEvent.class));
                    payState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<OrderEvent>("payState", OrderEvent.class));
                }
                
                @Override
                public void processElement(OrderEvent value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    
                    // 订单创建状态和支付状态都是null的时候, 表示这个订单的信息第一次进来
                    // 注册定时器
                    if(createState.value() == null && payState.value() == null){
                        ctx.timerService().registerEventTimeTimer(value.getEventTime() + 30 * 60 * 1000);
                    }else{
                        // 表示第二个进来了, 删除定时器
                        final long timer = (createState.value() != null
                            ? createState.value().getEventTime()
                            : payState.value().getEventTime()) + 30 * 60 * 1000;
                        ctx.timerService().deleteEventTimeTimer(timer);
                    }
                    
                    
                    final String eventType = value.getEventType();
                    
                    if ("create".equalsIgnoreCase(eventType)) { // 创建订单
                        if (payState.value() == null) { // 支付信息还没来
                            createState.update(value);
                        } else {
                            // 判断支付时间有没有超时
                            if (payState.value().getEventTime() - value.getEventTime() < 15 * 60 * 1000) {
                                out.collect("订单: " + value.getOrderId() + " 正常支付 ");
                            } else {
                                out.collect("订单: " + value.getOrderId() + " 在超时时间后支付, 检测系统是否有漏洞! ");
                            }
                        }
                        
                    } else {  // 支付
                        if(createState.value() == null){
                            // 支付来了, 但是订单创建还没有
                            payState.update(value);
                        }else{
                            // 创建已经到了, 支付又来了
                            // 判断支付时间有没有超时
                            if (value.getEventTime() - createState.value().getEventTime() < 15 * 60 * 1000) {
                                out.collect("订单: " + value.getOrderId() + " 正常支付 ");
                            } else {
                                out.collect("订单: " + value.getOrderId() + " 在超时时间后支付, 检测系统是否有漏洞! ");
                            }
                        }
                    }
                }
    
                @Override
                public void onTimer(long timestamp,
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    if(createState.value() != null){
                        // 有下单没有支付
                        out.collect("订单: " + createState.value().getOrderId() + " 有下单, 但是没有支付!");
                    }else{
                        // 没有下单, 有支付
                        out.collect("订单: " + payState.value().getOrderId() + " 没有下单, 但是有支付, 请检查系统是否设计缺陷!");
                    }
                }
            })
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
